package com.gjy.pulsar.disruptor;

import com.lmax.disruptor.EventHandler;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-08-06 11:05:35
 */
public class MessageBusinessHandler implements EventHandler<PulsarMessageEvent> {

    private static final Logger log = LoggerFactory.getLogger(MessageBusinessHandler.class);

    private Consumer<byte[]> consumer;

    public MessageBusinessHandler() {
    }

    public MessageBusinessHandler(Consumer<byte[]> consumer) {
        this.consumer = consumer;
    }

    @Override
    public void onEvent(PulsarMessageEvent event, long l, boolean b) throws Exception {
        Message<byte[]> msg = event.getPulsarMessage();
        try {
            // 1. 解析业务数据
            String content = new String(msg.getData());

            // 2. 执行业务逻辑（如订单处理）
            processBusiness(content);

            // 3. 消息ACK（确保处理成功后才确认）
            consumer.acknowledge(msg);
        } catch (Exception e) {
            consumer.negativeAcknowledge(msg); // 处理失败重试
        }
    }

    private void processBusiness(String content) {
        log.info("content: {}", content);
    }
}
